package com.hzh.flink.sink

import org.apache.flink.streaming.api.functions.sink.SinkFunction
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}

object Demo3SinkFunction {

    def main(args: Array[String]): Unit = {
      val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

      val linesDS: DataStream[String] = env.socketTextStream("master", 8888)

      //使用自定义sink
      linesDS.addSink(new SinkFunction[String] {
        /**
         * invoke 每一条数据都会执行一次
         *
         * @param value   ： 一条数据
         * @param context ： 上下文对象
         */
        override def invoke(value: String, context: SinkFunction.Context): Unit = {
          println(value)
        }
      })


      env.execute()
    }
}

